package net.lb.socketserver.bus;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import javax.annotation.Resource;
import net.lb.socketserver.handler.MessageAction;
import net.lb.socketserver.handler.MessageHandler;
import net.lb.socketserver.interceptor.MessageInterceptor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

@Resource
public class LocalHandlerCenter implements MessageHandlerCenter
{
    @Resource
    private List<MessageHandler>            handlerList;
    @Resource
    private List<MessageInterceptor>        interceptorList;
    private MessageAction                   messageAction;
    // 命令数组
    private byte[]                          command;
    // 命令数组的长度
    private int                             limit;
    private MessageHandler[]                handlers;
    private BlockingQueue<Message>          waitList     = new LinkedBlockingQueue<>();
    private int                             taskLimit    = Runtime.getRuntime().availableProcessors() * 2;
    private int                             runningTask  = 0;
    private ExecutorService                 threadPool   = null;
    private ExecutorCompletionService<Void> executorPool = null;
    private Logger                          logger       = LogManager.getLogger();
    
    @Override
    public void run()
    {
        try
        {
            while (true)
            {
                Message msg = waitList.take();
                byte msgCommand = msg.getCommand();
                int index = 0;
                for (; index < limit; index++)
                {
                    if (command[index] == msgCommand)
                    {
                        break;
                    }
                }
                if (index < limit)
                {
                    MessageHandler handler = handlers[index];
                    msg.setHandler(handler);
                    messageAction.addMessage(msg);
                    executorPool.submit(messageAction);
                    runningTask++;
                }
                else
                {
                    logger.error("不识别的命令类型，关闭通道");
                    msg.getChannelInfo().close();
                }
                while (executorPool.poll() != null)
                {
                    runningTask--;
                }
                if (runningTask >= taskLimit)
                {
                    executorPool.take();
                    runningTask--;
                }
            }
        }
        catch (InterruptedException e)
        {
            logger.info("{}被中断", Thread.currentThread().getName());
        }
        catch (Exception e)
        {
            logger.error("{}异常", Thread.currentThread().getName(), e);
        }
    }
    
    public void offerMessage(Message message)
    {
        waitList.offer(message);
    }
    
    public void stop()
    {
        threadPool.shutdownNow();
        try
        {
            threadPool.awaitTermination(50, TimeUnit.SECONDS);
        }
        catch (InterruptedException e)
        {
            throw new RuntimeException(e);
        }
        finally
        {
            
        }
    }
    
    @Override
    public void init()
    {
        MessageInterceptor[] interceptors = interceptorList.toArray(new MessageInterceptor[0]);
        handlers = handlerList.toArray(new MessageHandler[0]);
        messageAction = new MessageAction(interceptors);
        limit = handlers.length;
        command = new byte[limit];
        for (int i = 0; i < limit; i++)
        {
            command[i] = handlers[i].interestedDataPacketType();
        }
        threadPool = Executors.newFixedThreadPool(taskLimit, new ThreadFactory() {
            private int i = 1;
            
            @Override
            public Thread newThread(Runnable r)
            {
                Thread thread = new Thread(r, "业务处理线程-" + i);
                i++;
                return thread;
            }
        });
        executorPool = new ExecutorCompletionService<Void>(threadPool);
    }
    
    @Override
    public List<MessageHandler> getHandlerList()
    {
        return handlerList;
    }
    
    @Override
    public List<MessageInterceptor> getInterceptorList()
    {
        return interceptorList;
    }
    
    @Override
    public byte[] getCommand()
    {
        return command;
    }
}
